package tuyou.kafka.producer;

import java.time.Clock;
import java.util.Properties;
import java.util.Random;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * 消息生产者
 * 
 * @author：涂有
 * @date 2017年6月1日 上午11:27:47
 */
public class TimeStampProducer extends Thread {

	public static final String ip = "10.100.124.205:9092";
	private final KafkaProducer<String, String> producer;
	public static final String topic = "test2";

	public TimeStampProducer() {
		Properties properties = new Properties();
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ip);// 集群地址
		properties.put(ProducerConfig.CLIENT_ID_CONFIG, "MsgProducer");// 自定义客户端id
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringSerializer");// key
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringSerializer");// value
		// properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,CustomPartitioner.class.getCanonicalName());//自定义分区函数

		// properties.load("properties配置文件");
		this.producer = new KafkaProducer<String, String>(properties);
	}

	@Override
	public void run() {
		Clock clock = Clock.systemUTC();
		Random rand = new Random();
		while (true) {
			
			long lag = rand.nextInt(10000);
			String msg = (clock.millis() - lag) + "";
			producer.send(new ProducerRecord<>(topic, msg));
			System.out.println("发送：" + msg);
			try {
				Thread.sleep(lag);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public static void main(String args[]) {
		new TimeStampProducer().start();// 开始发送消息
	}
}